package top.lzzly.sync.binlog.binlog;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import top.lzzly.sync.binlog.model.BinlogDto;
import com.alibaba.fastjson.JSON;
import top.lzzly.sync.binlog.model.ChargeChannel;
import top.lzzly.sync.binlog.model.ChargeDevice;
import top.lzzly.sync.binlog.model.ChargeOrder;
import top.lzzly.sync.binlog.kafka.KafkaSender;

/**
 * @Author smq
 * @Date 2021/8/26 12:39
 * @Uint d9lab
 * @Description: maven简易集成cancel
 */
@Slf4j
@Component
@SuppressWarnings("InfiniteLoopStatement")
public class SimpleCanalClientExample implements CommandLineRunner {

    // kafka话题
    @Value("${spring.kafka.topic}")
    private String topic;

    // kafka分区
    @Value("${spring.kafka.partNum}")
    private int partNum;

    // Kafka备份数
    @Value("${spring.kafka.repeatNum}")
    private short repeatNum;

    // kafka地址
    @Value("${spring.kafka.bootstrap-servers}")
    private String kafkaHost;

    @Autowired
    private KafkaSender kafkaSender;

    @Async
    @Override
    public void run(String[] args) throws Exception {
        // 创建topic
        kafkaSender.createTopic(kafkaHost, topic, partNum, repeatNum);
        // 创建链接 47.115.23.102
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("47.115.23.102",
                11111), "example", "canal", "D9lab@2020");
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe("d9iot.charge_channel,d9iot.charge_order,d9iot.charge_device");
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId != -1 && size != 0) {
                    printEntry(message.getEntries());
                }
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } finally {
            connector.disconnect();
        }
    }

    private void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            EventType eventType = rowChage.getEventType();
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    String eventKey = entry.getHeader().getTableName() + ".delete";
                    printColumn(rowData.getBeforeColumnsList(), eventKey);
                } else if (eventType == EventType.INSERT) {
                    String eventKey = entry.getHeader().getTableName() + ".insert";
                    printColumn(rowData.getAfterColumnsList(), eventKey);
                } else {
                    String eventKey = entry.getHeader().getTableName() + ".update";
                    printColumn(rowData.getAfterColumnsList(), eventKey);
                }
            }
        }
    }

    private void printColumn(List<Column> columns, String eventKey) {
        Map<String, String> map = new ConcurrentHashMap<>();
        String msg = null;
        for (Column column : columns) {
            map.put(column.getName(), column.getValue());
        }
        if (eventKey.contains("charge_channel")) {
            ChargeChannel chargeChannel = ChargeChannel.builder()
                    .id(map.get("id"))
                    .devId(map.get("dev_id"))
                    .channelId(map.get("channel_id"))
                    .channelStatus(map.get("channel_status"))
                    .channelTime(map.get("channel_time"))
                    .channelPower(map.get("channel_power"))
                    .faultType(map.get("fault_type"))
                    .channelSignal(map.get("channel_signal"))
                    .costStandard(map.get("cost_standard"))
                    .gmtCreate(map.get("gmt_create"))
                    .gmtModified(map.get("gmt_modified"))
                    .isDelete(map.get("is_delete"))
                    .isSelect(null)
                    .channelVoltage(null)
                    .channelCurrent(null)
                    .selectCreate(null)
                    .build();
            msg = JSON.toJSONString(new BinlogDto(eventKey, chargeChannel));
        } else if (eventKey.contains("charge_order")){
            ChargeOrder chargeOrder = ChargeOrder.builder()
                    .id(map.get("id"))
                    .orderNo(map.get("order_no"))
                    .totalFee(map.get("total_fee"))
                    .devId(map.get("dev_id"))
                    .userId(Integer.parseInt(map.get("user_id")))
                    .channel(map.get("channel"))
                    .chargingTime(map.get("charging_time"))
                    .costStandard(map.get("cost_standard"))
                    .deviceId(map.get("device_id"))
                    .gmtCreate(map.get("gmt_create"))
                    .isDelete(map.get("is_delete"))
                    .location(map.get("location"))
                    .gmtModified(map.getOrDefault("gmt_modified", null))
                    .gmtChargeStatusModified(map.getOrDefault("gmt_charge_status_modified", null))
                    .refundFee(map.getOrDefault("refund_fee", null))
                    .refundId(map.getOrDefault("refund_id", null))
                    .refundNo(map.getOrDefault("refund_no", null))
                    .gmtRefund(map.getOrDefault("gmt_refund", null))
                    .transactionId(map.getOrDefault("transaction_id", null))
                    .build();
            msg = JSON.toJSONString(new BinlogDto(eventKey, chargeOrder));
        }else {
            ChargeDevice chargeDevice = ChargeDevice.builder()
                    .id(map.get("id"))
                    .communicationStatus(map.get("communication_status"))
                    .data(map.get("data"))
                    .devId(map.get("dev_id"))
                    .devName(map.get("dev_name"))
                    .devStatus(map.get("dev_status"))
                    .gmtCreate(map.get("gmt_create"))
                    .gmtModified(map.get("gmt_modified"))
                    .lat(map.get("lat"))
                    .log(null)
                    .isDelete(null)
                    .build();
            msg = JSON.toJSONString(new BinlogDto(eventKey, chargeDevice));
        }
        kafkaSender.send(topic, msg);
    }
}
